package handlers

import (
	"errors"
	"github.com/gin-gonic/gin"
	"github.com/gin-gonic/gin/binding"
	"github.com/jmoiron/sqlx"
	"learning/gooop/saga/mqs/database"
	"learning/gooop/saga/mqs/eventbus"
	"learning/gooop/saga/mqs/logger"
	"learning/gooop/saga/mqs/models"
	"learning/gooop/saga/mqs/models/events"
	"net/http"
	"time"
)

// Publish: accept msg post, save it, and publish to subscribers
func Publish(c *gin.Context) {
	// parse request
	msg := &models.TxMsg{}
	if err := c.ShouldBindBodyWith(&msg, binding.JSON); err != nil {
		c.AbortWithStatusJSON(
			http.StatusInternalServerError,
			gin.H{ "ok": false, "error": err.Error()})
		return
	}

	// fixme: validate msg

	// save to db
	lstInsertID := []int64{0}
	err := database.TX(func(db *sqlx.DB, tx *sqlx.Tx) error {
		id, e := saveTxMsg(db, tx, msg)
		if e != nil {
			return e
		}

		lstInsertID[0] = id
		return nil
	})

	// reply
	if err != nil {
		logger.Logf(
			"handlers.Publish, failed, msg=%s/%s/%s, err=%v",
			msg.GlobalID, msg.SubID, msg.Topic, err)
		c.JSON(http.StatusInternalServerError, gin.H{ "ok": false, "error": err.Error()})

	} else {
		c.JSON(http.StatusOK, gin.H { "ok": true })

		// deliver new msg at once
		lastInsertID := lstInsertID[0]
		if lastInsertID > 0 {
			logger.Logf("handlers.Publish, msg=%s/%s/%s, msgId=%d", msg.GlobalID, msg.SubID, msg.Topic, lastInsertID)
			go pubLiveMsg(lastInsertID)
		}
	}
}

// saveTxMsg: save msg, and copy to delivery_queue
func saveTxMsg(db *sqlx.DB, tx *sqlx.Tx, msg *models.TxMsg) (int64,error) {
	// insert ignore into tx_msg
	r, e := db.Exec(
		`insert or ignore into tx_msg(
			GlobalID, SubID, SenderID, 
			CreateTime, topic, content
		) values(?,?,?,?,?,?)`,
		msg.GlobalID, msg.SubID, msg.SenderID,
		msg.CreateTime, msg.Topic, msg.Content,
	)
	if e != nil {
		return 0,e
	}

	// if no rows affected, then ignore
	rowsAffected, e := r.RowsAffected()
	if e != nil {
		return 0, e
	}
	switch rowsAffected {
	case 0:
		return 0, nil
	case 1:
		break
	default:
		return 0, gOneRowsErr
	}

	// get last insert id
	lastInsertID,e := r.LastInsertId()
	if e != nil {
		return 0,e
	}

	if lastInsertID > 0 {
		// copy to delivery queue
		now := time.Now().UnixNano()
		r, e = db.Exec(`
			insert into delivery_queue(
				ClientID, NotifyUrl, 
				MsgId, GlobalID, SubID, SenderID, CreateTime, topic, content,  
				StatusFlag, UpdateTime) 
			select
				s.ClientID, s.NotifyURL,
			    t.id, t.GlobalID, t.SubID, t.SenderID, t.CreateTime, t.topic, t.content,
			    0, 0   
			from
				tx_msg t, subscriber s 
			where
			    t.id=?  
				and s.ExpireTime>?
				and s.topic=?
		`, lastInsertID, now, msg.Topic)
		if e != nil {
			return 0,e
		}
	}

	return lastInsertID, nil
}

// pubLiveMsg: using eventbus to notify delivery workers
func pubLiveMsg(msgId int64) {
	logger.Logf("handlers.Publish, pubLiveMsg %d", msgId)

	buf := [][]*models.QueuedMsg{ {} }
	err := database.DB(func(db *sqlx.DB) error {
		rows, e := db.Queryx("select * from delivery_queue where MsgId=?", msgId)
		if e != nil {
			return e
		}

		defer rows.Close()
		for rows.Next() {
			msg := &models.QueuedMsg{}
			e = rows.StructScan(msg)
			if e != nil {
				return e
			}

			buf[0] = append(buf[0], msg)
		}

		return nil
	})

	if err != nil {
		logger.Logf("handlers.Publish, pubLiveMsg, err=%v", err)
		return
	}

	rows := buf[0]
	logger.Logf("handlers.Publish, pubLiveMsg, msgId=%d, rows=%d", msgId, len(rows))
	for _,msg := range rows {
		logger.Logf("handlers.Publish, event=%s, clientID=%s, msg=%s/%s/%s",
			events.MsgPublishedEvent, msg.ClientID, msg.GlobalID, msg.SubID, msg.NotifyURL )
		eventbus.GlobalEventBus.Pub(events.MsgPublishedEvent, msg)
	}
}

var gOneRowsErr = errors.New("expecting one row affected")